unit tcpbuffers;

{$mode objfpc}{$H+}

interface

uses
  Classes, SysUtils, fpstructs, lNet, lNetSSL;

type

  TTCPSocket = TLSocket;

  { TTCPBuffer }

  TTCPBuffer = class ( TObject )
  private
    Fid: integer;      // this for debugging - probably not required otherwise
    FMessageLength: integer;
    fSocket: TLSocket;
    fConnection: TLTCP;
    fData: TMemoryStream;
    fheader: tfpHeader;
    FMessageComplete: Boolean;
    FOnReady: TNotifyEvent;
    function GetBytesToCome: integer;
    procedure SetHeader ( const AValue: tfpHeader ) ;
  public
    constructor Create( connection: TLTCP; aSocket: TLSocket );
    destructor Destroy; override;
    function ReadSocket ( Asocket: TLSocket ): Integer;
    function Connect(const Address: string; const APort: Word): Boolean;
    property Header: tfpHeader read fheader write SetHeader;
    property Data: TMemoryStream read fData;
    property Socket: TLSocket read fSocket;
    property OnReady: TNotifyEvent read FOnReady write FOnReady;
    property BytesToCome: integer read Getbytestocome;
    property MessageComplete: Boolean read FMessageComplete write FMessageComplete;
    property MessageLength: integer read FMessageLength write FMessageLength;
    property ID: integer read Fid;
  end;

  { TTCPSendBuffer }

  TTCPSendBuffer = class ( TTCPBuffer )
  public
    procedure AddData( txt: string );
    procedure AddData ( int: integer );
    procedure AddData ( dbl: Double );
    procedure AddData ( buf: TByteArray; buflen: integer );
    function SendData: Boolean;
  end;


  { TTCPConnection }

  TTCPConnection = class (TObject)
  private
    fConnection: TLTcp;
    fSession: TLSSLSession;
    fConnected: Boolean;
    FOnConnect: TLSocketEvent;
    FOnDisconnect: TLSocketEvent;
    FOnError: TLSocketErrorEvent;
    FOnReceiveMessage: TNotifyEvent;
    fReceiveBuffers: TList;
    fSendBuffers: TList;
    procedure DoCanSend(aSocket: TLSocket);
    procedure DoDisconnect(aSocket: TLSocket);
    procedure DoConnect(aSocket: TLSocket);
    procedure DoError(const msg: string; aSocket: TLSocket);
    procedure DoReceive(aSocket: TLSocket);
    function AwaitingData( aSocket: TLSocket ): Boolean;
    function GetUseSSL: Boolean;
    procedure SetUseSSL ( const AValue: Boolean ) ;
  public
    constructor Create;
    destructor Destroy; override;
    procedure SendMessage( mess: TTCPSendBuffer );
    procedure DeleteMessage( mess: TTCPBuffer ); //deletes a message from fReceiveBuffers
    procedure CallAction;
    function Connect(const Address: string; const APort: Word): Boolean;
    procedure Disconnect(const Forced: Boolean = False);
    property Connected: Boolean read fConnected;
    property Connection: TLTcp read fConnection write fConnection;
    property OnDisconnect: TLSocketEvent read FOnDisconnect write FOnDisconnect;
    property OnConnect: TLSocketEvent read FOnConnect write FOnConnect;
    property OnError: TLSocketErrorEvent read FOnError write FOnError;
    property OnReceiveMessage: TNotifyEvent read FOnReceiveMessage write FOnReceiveMessage;
    property UseSSL: Boolean read GetUseSSL write SetUseSSL;
  end;

  const
    SENDCHUNKSIZE = 1024;


implementation

{ TTCPBuffer }

function TTCPBuffer.Getbytestocome: integer;
begin
  Result := FMessageLength-fData.Size;
end;

procedure TTCPBuffer.SetHeader ( const AValue: tfpHeader ) ;
begin
  fheader := AValue;
  fData.Position := 0;
  fData.Write(fheader, SizeOf(fheader));
  fData.Seek(0, soFromBeginning);
end;

constructor TTCPBuffer.Create ( connection: TLTCP; aSocket: TLSocket ) ;
begin
  inherited Create;
  Fid := Random(10000)+1;
  FMessageComplete := false;
  FMessageLength := 0;
  fConnection := connection;
  fSocket := aSocket;
  fData := TMemoryStream.Create;
  {$IFDEF DEBUGTCPBUF} writeln('buffer ', fid, ' created'); {$ENDIF}
end;

destructor TTCPBuffer.Destroy;
begin
  fdata.Free;
  {$IFDEF DEBUGTCPBUF}
  writeln('Buffer ', fid, ' freed');
  writeln('');
{$ENDIF}
  inherited Destroy;
end;

function TTCPBuffer.ReadSocket ( Asocket: TLSocket ) : Integer;
var
  buf: array of byte;
begin
  Result := -1;
  {$IFDEF DEBUGTCPBUF} writeln('Buffer ', FID, ' ReadSocket: bytestocome=',BytesToCome);   {$ENDIF}
  SetLength(buf, BytesToCome);
  if BytesToCome>0
     then Result := fConnection.Get(buf[0], BytesToCome, Asocket);
  {$IFDEF DEBUGTCPBUF} writeln('Read ', Result, ' bytes'); {$ENDIF}
  fdata.Seek(0, soFromEnd);
  if Result>-1 then
     fdata.Write(buf[0], Result);
  if fData.Size=FMessageLength
      then if Assigned(FOnReady) then
        begin
          //TODO: implement MD5 check ?
           {checkMD5 := MD5String(fDatabuffer);
           if MD5Match( checkMD5, fCurrentHeader.md5 )
              then writeln('MD5 is OK')
              else writeln('MD5 differs'); }
          FMessageComplete := true;
          fData.Position := SizeOf(fheader);
          try
          FOnReady(Self);
          except
            on E: Exception do
               writeln('Error when processing message data: ', E.Message);
          end;
        end
  else Raise Exception.Create('FonReady not assigned');
end;

function TTCPBuffer.Connect ( const Address: string; const APort: Word
  ) : Boolean;
begin
  Result := fConnection.Connect(Address, APort);
end;

{ TTCPSendBuffer }

procedure TTCPSendBuffer.AddData ( txt: string ) ;
begin
  fData.Seek(0, soFromEnd);
  if Length(txt)>0
     then fData.Write(txt[1], Length(txt));
  fdata.Seek(0, soFromBeginning);
  fheader.datalength := fData.Size-SizeOf(fheader);
  fData.Write(fheader, SizeOf(fheader));
  fdata.Seek(0, soFromBeginning);
end;

procedure TTCPSendBuffer.AddData ( int: integer ) ;
begin
  fData.Seek(0, soFromEnd);
  fData.Write(int, SizeOf(int));
  fdata.Seek(0, soFromBeginning);
  fheader.datalength := fData.Size-SizeOf(fheader);
  fData.Write(fheader, SizeOf(fheader));
  fdata.Seek(0, soFromBeginning);
end;

procedure TTCPSendBuffer.AddData(dbl: Double);
begin
  fData.Seek(0, soFromEnd);
  fData.Write(dbl, SizeOf(dbl));
  fdata.Seek(0, soFromBeginning);
  fheader.datalength := fData.Size-SizeOf(fheader);
  fData.Write(fheader, SizeOf(fheader));
  fdata.Seek(0, soFromBeginning);
end;

procedure TTCPSendBuffer.AddData ( buf: TByteArray; buflen: integer ) ;
begin
  fData.Seek(0, soFromEnd);
  fData.Write(buf[0], buflen);
  fdata.Seek(0, soFromBeginning);
  fheader.datalength := fData.Size-SizeOf(fheader);
  fData.Write(fheader, SizeOf(fheader));
  fdata.Seek(0, soFromBeginning);
end;

function TTCPSendBuffer.SendData: Boolean;
var
  P: Int64;
  buffer: array of byte;
  b: Int64;
  sent: LongInt;
begin
  p := fData.Position;
  repeat
  b := SENDCHUNKSIZE;
  if b > fData.Size-p
     then b := fData.Size-p;
  SetLength(buffer, b);
  fData.Read(buffer[0], b);
  sent := fConnection.Send(buffer[0], b, fSocket);
  p := p + sent;
  fData.Position := p;
{$IFDEF DEBUGTCPBUF} WriteLn('Buffer ', FID, ' Sent ', sent,  ' bytes : p=', p, ' Data.Size=', fData.size); {$ENDIF}
  until (Sent=0) or (fData.Size <= p);

  if fData.Size <= p then
    begin
      {$IFDEF DEBUGTCPBUF} writeln('Buffer ', FID, ' successfully sent cached message (', fData.Size, ' bytes)'); {$ENDIF}
      FMessageComplete := true;
      Result := true;
    end
  else Result := false; // not finished sending data
end;


{ TTCPConnection }

procedure TTCPConnection.DoCanSend ( aSocket: TLSocket ) ;
var
  x: Integer;
begin
  x := fSendBuffers.Count-1;
  {$IFDEF DEBUGTCPBUF} writeln ('TTCPConnection.DoCanSend');
  if aSocket<>nil then writeln(aSocket.PeerAddress);       {$ENDIF}
  while (x>-1) do
        begin
          if TTCPSendBuffer(fSendBuffers[x]).Socket=aSocket then
            begin
              {$IFDEF DEBUGTCPBUF} writeln('buffer at ', x);    {$ENDIF}
              TTCPSendBuffer(fSendBuffers[x]).SendData;
              if TTCPSendBuffer(fSendBuffers[x]).MessageComplete then
                    begin
                      TTCPSendBuffer(fSendBuffers[x]).Free;
                      fSendBuffers.Delete(x);
                    end
              else x := -1;
            end;
          Dec(x);
        end;
end;

procedure TTCPConnection.DoDisconnect ( aSocket: TLSocket ) ;
var
  x: Integer;
begin
  fConnected := false;
  Writeln('Lost connection');
  if asocket<>nil then writeln(aSocket.PeerAddress); // write info if connection was lost
  For x := fReceiveBuffers.Count-1 downto 0 do
      if TTCPBuffer(fReceiveBuffers[x]).Socket=aSocket then
        begin
          if not TTCPBuffer(fReceiveBuffers[x]).MessageComplete
             then writeln('Discarding unsent message to ', aSocket.PeerAddress, ' (', TTCPBuffer(fReceiveBuffers[x]).Data.Size, ' bytes)');
          TTCPBuffer(fReceiveBuffers[x]).Free;
          fReceiveBuffers.Delete(x);
        end;
  if Assigned(FOnDisconnect)
    then FOnDisconnect(aSocket);
end;

procedure TTCPConnection.DoConnect ( aSocket: TLSocket ) ;
begin
  fConnected := true;
  if Assigned (FOnConnect)
     then FOnConnect(aSocket);
end;

procedure TTCPConnection.DoError (const msg: string; aSocket: TLSocket);
begin
  {$IFDEF DEBUGTCPBUF}
  writeln(msg, ' (connection to ', aSocket.PeerAddress, ')');
  {$ENDIF}
  if Assigned (FOnError)
     then FOnError(msg, aSocket);
end;

procedure TTCPConnection.DoReceive ( aSocket: TLSocket ) ;
var
  header: tfpHeader;
  bytesreceived: LongInt;
  abuffer: TTCPBuffer;
  x: Integer;
begin
  writeln('incoming data... ', aSocket.PeerAddress);
  if not AwaitingData(aSocket) then
    begin
      bytesreceived := fConnection.Get(header, SizeOf(header), aSocket);
      writeln('bytesreceived=', bytesreceived);
      if bytesreceived=SizeOf(header) then
         begin
           if header.ID=HEADER_ID then // we have a new message
                 begin
                   abuffer := TTCPBuffer.Create(fConnection, aSocket);
                   abuffer.Header := header;
                   abuffer.MessageLength := SizeOf(header)+header.datalength;
      //             writeln('messagelength is ', abuffer.MessageLength);
                   abuffer.OnReady := FOnReceiveMessage;
                   fReceiveBuffers.Add(abuffer);
                   {$IFDEF DEBUGTCPBUF} writeln('Message stored at ', fReceiveBuffers.IndexOf(abuffer));  {$ENDIF}
                   abuffer.ReadSocket(aSocket);
                   fConnection.CallAction;
                 end
            else begin
                   writeln('invalid header ', header.ID);
                 end;
         end
      else writeln('Short header');
    end
  else
    begin
      writeln('Partial message detected');
      x := fReceiveBuffers.Count-1;
      while (x>-1) and (TTCPBuffer( fReceiveBuffers[x] ).Socket<>aSocket)
          do Dec(x);
      if x<0 then
            begin
              writeln('TCP Receive error: Buffer not found for socket');
              exit;
            end
            else TTCPBuffer( fReceiveBuffers[x] ).ReadSocket(aSocket);
    end;
end;

function TTCPConnection.AwaitingData ( aSocket: TLSocket ) : Boolean;
var
  x: Integer;
begin
  Result := False;
  x := fReceiveBuffers.Count-1;
  while (x > -1) and ( TTCPBuffer(fReceiveBuffers[x]).Socket <> aSocket )
        do begin
          write ('found socket for ', TTCPBuffer(fReceiveBuffers[x]).Socket.PeerAddress, ' at ', x);
          if TTCPBuffer(fReceiveBuffers[x]).MessageComplete
             then writeln(' (processed)')
             else writeln(' (unprocessed)');
          Dec(x);
        end;
  if x > -1
     then Result := not TTCPBuffer(fReceiveBuffers[x]).MessageComplete
     else writeln('no buffer found for ', aSocket.PeerAddress);
end;

function TTCPConnection.GetUseSSL: Boolean;
begin
  Result := fSession.SSLActive;
end;

procedure TTCPConnection.SetUseSSL ( const AValue: Boolean ) ;
begin
  if AValue<>fSession.SSLActive
     then fSession.SSLActive := AValue;
end;

constructor TTCPConnection.Create;
begin
  fSendBuffers := Tlist.Create;
  fReceiveBuffers := Tlist.Create;
  fConnection := TLTcp.Create(nil);
  fSession := TLSSLSession.Create(nil);
  fSession.Method := msSSLv2or3;
  fSession.SSLActive := false;
  fConnection.Session := fSession;
  with fConnection do
       begin
         OnCanSend := @DoCanSend;
         OnConnect := @DoConnect;
         OnDisconnect := @DoDisconnect;
         OnError := @DoError;
         OnReceive := @DoReceive;
       end;
end;

destructor TTCPConnection.Destroy;
var
  x: Integer;
begin
  if fConnected
     then fConnection.Disconnect(true);
  fConnection.Free;  //TODO: need to disconnect politely
  for x := fSendBuffers.Count-1 downto 0 do
        TTCPSendBuffer(fSendBuffers[x]).Free;
  fSendBuffers.Free;
  for x := fReceiveBuffers.Count-1 downto 0 do
        TTCPBuffer(fReceiveBuffers[x]).Free;
  fSession.Free;
  fReceiveBuffers.Free;
  inherited Destroy;
end;

procedure TTCPConnection.SendMessage ( mess: TTCPSendBuffer ) ;
begin
  mess.fConnection := fConnection;
  fSendBuffers.Insert(0, mess); // DoCansend looks for the highest index
  {$IFDEF DEBUGTCPBUF} writeln('Inserted at ', fSendBuffers.IndexOf(mess));   {$ENDIF}
  DoCanSend(mess.Socket);
end;

procedure TTCPConnection.DeleteMessage ( mess: TTCPBuffer ) ;
var
  x: Integer;
begin
  x := 0;
  while ( x<fReceiveBuffers.Count ) and ( TTCPBuffer(fReceiveBuffers[x])<>mess )
        do Inc(x);
  if x<fReceiveBuffers.Count then
        begin
          {$IFDEF DEBUGTCPBUF} WriteLn('Deleting old message from Terminal #', TTCPBuffer(fReceiveBuffers[x]).Header.TerminalID);    {$ENDIF}
          TTCPBuffer(fReceiveBuffers[x]).Free;
          fReceiveBuffers.Delete(x);
        end;
end;

procedure TTCPConnection.CallAction;
begin
  //write('.');
  fConnection.CallAction;
end;

function TTCPConnection.Connect(const Address: string; const APort: Word
  ): Boolean;
begin
  Result := fConnection.Connect(Address, APort);
end;

procedure TTCPConnection.Disconnect(const Forced: Boolean);
begin
  fConnection.Disconnect(Forced);
end;

end.

